package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.commons.lang.time.FastDateFormat;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author: xu
 * @desc: 用户行为日志的DWD层
 * 数据流：web/app -> Nginx -> SpringBoot采集 -> Kafka(ods) -> FlinkApplication -> Kafka(dwd)
 * 程  序：mockLog -> Nginx -> Logger.sh -> Kafka(ZK) -> BaseLogApp -> kafka
 */
public class BaseLogApp {
    private static final String TOPIC_PAGE = "dwd_page_log";
    private static final String TOPIC_START = "dwd_start_log";
    private static final String TOPIC_DISPLAY = "dwd_display_log";

    public static void main(String[] args) throws Exception {
        // TODO 1.准备环境
        // 1.1 创建Flink流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(1);
        // 1.3设置Checkpoint，每60000ms开始一次checkpoint，模式是EXACTLY_ONCE（默认）
        // env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        // env.getCheckpointConfig().setCheckpointTimeout(60000);
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // env.setStateBackend(new FsStateBackend("hdfs://node1:8020/gmall/checkpoint/BaseLogApp"));
        // System.setProperty("HADOOP_USER_NAME","root");

        // TODO 2.从Kafka中读取数据
        String topic = "ods_base_log";
        String groupId = "base_log_app_group";
        // 2.1 调用Kafka工具类，获取FlinkKafkaConsumer的数据
        DataStreamSource<String> kafkaStream = env.addSource(MyKafkaUtil.getKafkaSource(topic, groupId));

        // TODO 3.对读取到的数据格式进行转换，String -> json
        SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaStream.map(JSON::parseObject);
        // jsonObjStream.print("json>>>");

        /*
          TODO 4.识别新老访客，前端也会对新老状态进行记录，有可能会不准，这里是再次做一个确认
           保存mid某天方法情况（将首次访问日期作为状态保存起来），等后面该设备在有日志过来的时候，从状态中获取日期
           和日志产生日志进行对比。如果状态不为空，并且状态日期和当前日期不相等，说明是老访客，如果is_new标记是1，那么对其状态进行修复
         */
        SingleOutputStreamOperator<JSONObject> jsonStreamWithFlag = jsonObjStream
                // 4.1 根据mid对日志进行分组
                .keyBy(data -> data.getJSONObject("common").getString("mid"))
                // 4.2 新老方法状态修复，状态分为算子和键控状态，我们这里要记录某一个设备的访问，使用键控状态比较合适
                .map(new RichMapFunction<JSONObject, JSONObject>() {
                    // 定义该mid访问状态
                    private transient ValueState<String> lastVisitDateState;
                    // 定义日期格式化对象
                    private FastDateFormat fdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 对状态以及日期格式进行初始化
                        lastVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<>("last-visit-date-state", String.class));
                        fdf = FastDateFormat.getInstance("yyyyMMdd");
                    }

                    @Override
                    public JSONObject map(JSONObject jsonObj) throws Exception {
                        // 获取当前日志标记状态
                        String isNew = jsonObj.getJSONObject("common").getString("is_new");
                        // 如果是新访客则修复
                        if ("1".equals(isNew)) {
                            // 获取上次当前mid对象的状态
                            String lastVisitDate  = lastVisitDateState.value();
                            // 对当前日志的日期格式
                            String curVisitDate = fdf.format(jsonObj.getLong("ts"));
                            // 如果状态不为空，并且状态日期和当前日期不相等，说明是老访客
                            if (lastVisitDate != null && lastVisitDate.length() != 0) {
                                // 判断是否为同一天数据
                                if (!lastVisitDate.equals(curVisitDate)) {
                                    jsonObj.getJSONObject("common").put("is_new", "0");
                                }
                            } else {
                                // 表示是第一次访问，将当前访问日志作为状态值
                                lastVisitDateState.update(curVisitDate);
                            }
                        }
                        return jsonObj;
                    }
                });
        // jsonStreamWithFlag.print(">>>");

        /*
          TODO 5.分流，根据日志数据内容，将日志数据分为3类,页面日志、启动日志、曝光日志。
           页面日志输出到主流，启动和曝光日志输出到侧输出流
           侧输出流：1.接收迟到数据，2.分流
         */
        // 定义启动和曝光侧输出流标签
        OutputTag<String> startTag = new OutputTag<>("start", TypeInformation.of(String.class));
        OutputTag<String> displayTag = new OutputTag<>("display", TypeInformation.of(String.class));

        SingleOutputStreamOperator<String> pageStream = jsonStreamWithFlag
                .process(new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<String> out) throws Exception {
                        // 获取启动日志标记
                        JSONObject startJsonObj = jsonObj.getJSONObject("start");
                        // 判断是否为启动日志
                        if (startJsonObj != null && startJsonObj.size() > 0) {
                            // 如果是启动日志，输出到启动侧输出流，将json格式转换为字符串，方便向侧输出流输出以及向kafka中写入
                            ctx.output(startTag, jsonObj.toJSONString());
                        } else {
                            // 如果不是启动日志，说明是页面日志，输出到主流
                            out.collect(jsonObj.toJSONString());
                            // 如果不是启动日志，获取曝光日志标记（曝光日志中也携带了页面）
                            JSONArray displays = jsonObj.getJSONArray("displays");
                            // 判断是否为曝光日志
                            if (displays != null && displays.size() > 0) {
                                // 获取页面id
                                String pageId = jsonObj.getJSONObject("page").getString("page_id");
                                // 如果是曝光日志，遍历输出到侧输出流
                                for (int i = 0; i < displays.size(); i++) {
                                    // 获取每一条曝光事件
                                    JSONObject displaysJsonObj = displays.getJSONObject(i);
                                    // 给每一条曝光事件加pageId
                                    displaysJsonObj.put("page_id", pageId);
                                    ctx.output(displayTag, displaysJsonObj.toString());
                                }
                            }
                        }
                    }
                });
        // 获取侧输出流
        DataStream<String> startStream = pageStream.getSideOutput(startTag);
        DataStream<String> displayStream = pageStream.getSideOutput(displayTag);

        // 打印输出
        pageStream.print("page>>>");
        startStream.print("start>>>");
        displayStream.print("display>>>");

        // TODO 6.将不同流的数据写回到kafka的不同topic中
        pageStream.addSink(MyKafkaUtil.getKafkaSink(TOPIC_PAGE));
        startStream.addSink(MyKafkaUtil.getKafkaSink(TOPIC_START));
        displayStream.addSink(MyKafkaUtil.getKafkaSink(TOPIC_DISPLAY));

        env.execute(BaseLogApp.class.getSimpleName());
    }
}